package com.atguigu.hbase;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.hbase.Cell;
import org.apache.hadoop.hbase.CellUtil;
import org.apache.hadoop.hbase.CompareOperator;
import org.apache.hadoop.hbase.TableName;
import org.apache.hadoop.hbase.client.*;
import org.apache.hadoop.hbase.filter.ColumnValueFilter;
import org.apache.hadoop.hbase.filter.FilterList;
import org.apache.hadoop.hbase.filter.SingleColumnValueFilter;
import org.apache.hadoop.hbase.util.Bytes;

import java.io.IOException;
import java.util.List;

/**
 * @author Felix
 * @date 2024/1/19
 * 操作Hbase的API介绍
 */
public class HbaseDemo {
    //获取hbase连接对象
    public static Connection getHbaseConnection(){
        Configuration config = new Configuration();
        config.set("hbase.zookeeper.quorum","hadoop102,hadoop103,hadoop104");

        try {
            Connection connection = ConnectionFactory.createConnection(config);
            // Connection connection = ConnectionFactory.createConnection();
            return connection;
        } catch (IOException e) {
            throw new RuntimeException();
        }
    }
    //关闭hbase连接对象
    public static void closeHbaseConnection(Connection conn){
        if(conn != null && !conn.isClosed()){
            try {
                conn.close();
            } catch (IOException e) {
                e.printStackTrace();
            }
        }
    }

    /**
     * 建表
     * @param conn          连接对象
     * @param namespace     命名空间
     * @param tableName     表名
     * @param families      列族
     */
    public static void createTable(Connection conn,String namespace,String tableName,String... families){
        if(families.length < 1){
            System.out.println("~~建表至少需要一个列族~~");
            return;
        }
        //获取Admin对象   自动资源释放
        try (Admin admin = conn.getAdmin()){
            //创建TableName对象
            TableName tableNameObj = TableName.valueOf(namespace,tableName);
            if (admin.tableExists(tableNameObj)){
                System.out.println("~~要创建的表已经存在~~");
                return ;
            }
            TableDescriptorBuilder tableDescriptorBuilder = TableDescriptorBuilder.newBuilder(tableNameObj);

            for (String family : families) {
                ColumnFamilyDescriptorBuilder columnFamilyDescriptorBuilder = ColumnFamilyDescriptorBuilder.newBuilder(Bytes.toBytes(family));
                tableDescriptorBuilder.setColumnFamily(columnFamilyDescriptorBuilder.build());
            }

            admin.createTable(tableDescriptorBuilder.build());
        } catch (IOException e) {
            e.printStackTrace();
        }
    }

    //删表
    public static void dropTable(Connection conn,String namespace,String tableName){
        try (Admin admin = conn.getAdmin()){
            TableName tableNameObj = TableName.valueOf(namespace, tableName);
            if(!admin.tableExists(tableNameObj)){
                System.out.println("要删除的表不存在");
                return;
            }
            admin.disableTable(tableNameObj);
            admin.deleteTable(tableNameObj);
        } catch (IOException e) {
            e.printStackTrace();
        }
    }

    /**
     * 向表中插入数据
     * @param conn          连接对象
     * @param namespace     表空间
     * @param tableName     表名
     * @param rowKey        rowkey
     * @param falimyName    列族名
     * @param columnName    列名
     * @param cellValue     值
     */
    public static void putCell(Connection conn,String namespace,String tableName,String rowKey,
                               String falimyName,String columnName,String cellValue){
        TableName tableNameObj = TableName.valueOf(namespace, tableName);
        try (Table table = conn.getTable(tableNameObj)){
            Put put = new Put(Bytes.toBytes(rowKey));
            put.addColumn(Bytes.toBytes(falimyName),Bytes.toBytes(columnName),Bytes.toBytes(cellValue));
            table.put(put);
        } catch (IOException e) {
            e.printStackTrace();
        }
    }

    //从表中插入数据---get
    public static void getRow(Connection conn,String namespace,String tableName,String rowKey){
        TableName tableNameObj = TableName.valueOf(namespace, tableName);
        try (Table table = conn.getTable(tableNameObj)){
            Get get = new Get(Bytes.toBytes(rowKey));
            //查询所有版本
            get.readAllVersions();

            Result result = table.get(get);
            //注意：如果一个rowkey对应多个cell，这种方式不合适
            // String resStr = Bytes.toString(result.value());
            // System.out.println(resStr);
            //如果要获取所有的cell
            List<Cell> cells = result.listCells();
            for (Cell cell : cells) {
                //通过CellUtil工具类从cell对象中获取对应的rowkey，列族、列名以及列值信息
                System.out.println(Bytes.toString(CellUtil.cloneRow(cell)));
                System.out.println(Bytes.toString(CellUtil.cloneFamily(cell)));
                System.out.println(Bytes.toString(CellUtil.cloneQualifier(cell)));
                System.out.println(Bytes.toString(CellUtil.cloneValue(cell)));
                System.out.println("~~~~~~~");
            }

        } catch (IOException e) {
            e.printStackTrace();
        }
    }

    //通过scan获取表中多条数据
    public static void scanRow(Connection conn,String namespace,String tableName,String startRow,String stopRow){
        TableName tableNameObj = TableName.valueOf(namespace, tableName);
        try (Table table = conn.getTable(tableNameObj)){
            Scan scan = new Scan();
            scan.withStartRow(Bytes.toBytes(startRow));
            scan.withStopRow(Bytes.toBytes(stopRow),true);
            ResultScanner scanner = table.getScanner(scan);
            for (Result result : scanner) {
                List<Cell> cells = result.listCells();
                for (Cell cell : cells) {
                    System.out.println(Bytes.toString(CellUtil.cloneRow(cell)));
                    System.out.println(Bytes.toString(CellUtil.cloneFamily(cell)));
                    System.out.println(Bytes.toString(CellUtil.cloneQualifier(cell)));
                    System.out.println(Bytes.toString(CellUtil.cloneValue(cell)));
                }
                System.out.println("~~~~~~~~~~~~");
            }
        } catch (IOException e) {
            e.printStackTrace();
        }
    }

    //带过滤的扫描
    public static void scanRowByFilter(Connection conn,String namespace,String tableName,String startRow,String stopRow,
                                       String columnFamily,String columnName,String cellValue){
        TableName tableNameObj = TableName.valueOf(namespace, tableName);
        try (Table table = conn.getTable(tableNameObj)){
            Scan scan = new Scan();
            scan.withStartRow(Bytes.toBytes(startRow));
            scan.withStopRow(Bytes.toBytes(stopRow),true);

            //列值过滤器   过滤出单个Cell
            ColumnValueFilter columnValueFilter = new ColumnValueFilter(
                Bytes.toBytes(columnFamily),
                Bytes.toBytes(columnName),
                CompareOperator.EQUAL,
                Bytes.toBytes(cellValue)
            );
            // scan.setFilter(columnValueFilter);

            //单列值过滤器  过滤出符合条件的当前rowkey对应的所有cell
            //注意： 如果表中的一行数据没有查询的列，也会将这行数据查询出来
            SingleColumnValueFilter singleColumnValueFilter = new SingleColumnValueFilter(
                Bytes.toBytes(columnFamily),
                Bytes.toBytes(columnName),
                CompareOperator.EQUAL,
                Bytes.toBytes(cellValue)
            );
            // scan.setFilter(singleColumnValueFilter);

            FilterList filterList = new FilterList();
            filterList.addFilter(columnValueFilter);
            filterList.addFilter(singleColumnValueFilter);
            scan.setFilter(filterList);

            ResultScanner scanner = table.getScanner(scan);
            for (Result result : scanner) {
                List<Cell> cells = result.listCells();
                for (Cell cell : cells) {
                    System.out.println(Bytes.toString(CellUtil.cloneRow(cell)));
                    System.out.println(Bytes.toString(CellUtil.cloneFamily(cell)));
                    System.out.println(Bytes.toString(CellUtil.cloneQualifier(cell)));
                    System.out.println(Bytes.toString(CellUtil.cloneValue(cell)));
                    System.out.println("~~~~~~~~~~~~");
                }
            }
        } catch (IOException e) {
            e.printStackTrace();
        }
    }

    //删除数据  作业

    public static void main(String[] args) {
        Connection conn = getHbaseConnection();
        // createTable(conn,"bigdata","student2","info","msg");
        // dropTable(conn,"bigdata","student2");
        // putCell(conn,"bigdata","student","1001","info"
        //     ,"name","zhangsanfeng");
        // putCell(conn,"bigdata","student","1001","info"
        //     ,"age","18");


        // getRow(conn,"bigdata","student","1001");
        // scanRow(conn,"bigdata","student","1001","1003");
        scanRowByFilter(conn,"bigdata","student","1001","1005",
            "info","name111","lisi");
        closeHbaseConnection(conn);
    }

}
